// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/dwio/common/selective_column_reader_internal.h>

namespace kumo::pollux::dwio::common {
    class SelectiveByteRleColumnReader : public SelectiveColumnReader {
    public:
        SelectiveByteRleColumnReader(
            const TypePtr &requestedType,
            std::shared_ptr<const dwio::common::TypeWithId> fileType,
            dwio::common::FormatParams &params,
            pollux::common::ScanSpec &scanSpec)
            : SelectiveColumnReader(
                requestedType,
                std::move(fileType),
                params,
                scanSpec) {
        }

        bool hasBulkPath() const override {
            return false;
        }

        void getValues(const RowSet &rows, VectorPtr *result) override;

        template<
            typename Reader,
            bool isDense,
            bool kEncodingHasNulls,
            typename ExtractValues>
        void processFilter(
            pollux::common::Filter *filter,
            ExtractValues extractValues,
            const RowSet &rows);

        template<typename Reader, bool isDense>
        void processValueHook(const RowSet &rows, ValueHook *hook);

        template<
            typename Reader,
            typename TFilter,
            bool isDense,
            typename ExtractValues>
        void readHelper(
            pollux::common::Filter *filter,
            const RowSet &rows,
            ExtractValues extractValues);

        template<typename Reader, bool kEncodingHasNulls>
        void
        readCommon(int64_t offset, const RowSet &rows, const uint64_t *incomingNulls);
    };

    template<
        typename Reader,
        typename TFilter,
        bool isDense,
        typename ExtractValues>
    void SelectiveByteRleColumnReader::readHelper(
        pollux::common::Filter *filter,
        const RowSet &rows,
        ExtractValues extractValues) {
        reinterpret_cast<Reader *>(this)->readWithVisitor(
            rows,
            ColumnVisitor<int8_t, TFilter, ExtractValues, isDense>(
                *reinterpret_cast<TFilter *>(filter), this, rows, extractValues));
    }

    template<
        typename Reader,
        bool isDense,
        bool kEncodingHasNulls,
        typename ExtractValues>
    void SelectiveByteRleColumnReader::processFilter(
        pollux::common::Filter *filter,
        ExtractValues extractValues,
        const RowSet &rows) {
        using pollux::common::FilterKind;
        switch (filter ? filter->kind() : FilterKind::kAlwaysTrue) {
            case FilterKind::kAlwaysTrue:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    filter, rows, extractValues);
                break;
            case FilterKind::kIsNull:
                if constexpr (kEncodingHasNulls) {
                    filterNulls<int8_t>(
                        rows,
                        true,
                        !std::is_same_v<decltype(extractValues), dwio::common::DropValues>);
                } else {
                    readHelper<Reader, pollux::common::IsNull, isDense>(
                        filter, rows, extractValues);
                }
                break;
            case FilterKind::kIsNotNull:
                if constexpr (
                    kEncodingHasNulls &&
                    std::is_same_v<decltype(extractValues), dwio::common::DropValues>) {
                    filterNulls<int8_t>(rows, false, false);
                } else {
                    readHelper<Reader, pollux::common::IsNotNull, isDense>(
                        filter, rows, extractValues);
                }
                break;
            case FilterKind::kBigintRange:
                readHelper<Reader, pollux::common::BigintRange, isDense>(
                    filter, rows, extractValues);
                break;
            case FilterKind::kNegatedBigintRange:
                readHelper<Reader, pollux::common::NegatedBigintRange, isDense>(
                    filter, rows, extractValues);
                break;
            case FilterKind::kBigintValuesUsingBitmask:
                readHelper<Reader, pollux::common::BigintValuesUsingBitmask, isDense>(
                    filter, rows, extractValues);
                break;
            case FilterKind::kNegatedBigintValuesUsingBitmask:
                readHelper<
                    Reader,
                    pollux::common::NegatedBigintValuesUsingBitmask,
                    isDense>(filter, rows, extractValues);
                break;
            default:
                readHelper<Reader, pollux::common::Filter, isDense>(
                    filter, rows, extractValues);
                break;
        }
    }

    template<typename Reader, bool isDense>
    void SelectiveByteRleColumnReader::processValueHook(
        const RowSet &rows,
        ValueHook *hook) {
        using namespace kumo::pollux::aggregate;
        switch (hook->kind()) {
            case aggregate::AggregationHook::kBigintSum:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(), rows, ExtractToHook<SumHook<int64_t> >(hook));
                break;
            default:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(), rows, ExtractToGenericHook(hook));
        }
    }

    template<typename Reader, bool kEncodingHasNulls>
    void SelectiveByteRleColumnReader::readCommon(
        int64_t offset,
        const RowSet &rows,
        const uint64_t *incomingNulls) {
        prepareRead<int8_t>(offset, rows, incomingNulls);
        const bool isDense = rows.back() == rows.size() - 1;
        pollux::common::Filter *filter =
                scanSpec_->filter() ? scanSpec_->filter() : &dwio::common::alwaysTrue();
        if (scanSpec_->keepValues()) {
            if (scanSpec_->valueHook()) {
                if (isDense) {
                    processValueHook<Reader, true>(rows, scanSpec_->valueHook());
                } else {
                    processValueHook<Reader, false>(rows, scanSpec_->valueHook());
                }
                return;
            }
            if (isDense) {
                processFilter<Reader, true, kEncodingHasNulls>(
                    filter, dwio::common::ExtractToReader(this), rows);
            } else {
                processFilter<Reader, false, kEncodingHasNulls>(
                    filter, dwio::common::ExtractToReader(this), rows);
            }
        } else {
            if (isDense) {
                processFilter<Reader, true, kEncodingHasNulls>(
                    filter, dwio::common::DropValues(), rows);
            } else {
                processFilter<Reader, false, kEncodingHasNulls>(
                    filter, dwio::common::DropValues(), rows);
            }
        }
    }
} // namespace kumo::pollux::dwio::common
